#include "boost/asio/io_context.hpp"
#include "boost/cobalt/detached.hpp"
#include "boost/cobalt/task.hpp"
#include <atomic>
#include <boost/asio/detached.hpp>
#include <boost/asio/post.hpp>
#include <boost/asio/use_future.hpp>
#include <boost/cobalt.hpp>
#include <boost/cobalt/run.hpp>
#include <boost/cobalt/spawn.hpp>
#include <boost/cobalt/this_thread.hpp>
#include <boost/cobalt/thread.hpp>
#include <boost/lockfree/spsc_queue.hpp>
#include <chrono>
#include <cstddef>
#include <curl/curl.h>
#include <iostream>
#include <thread>

namespace cobalt = boost::cobalt;

struct request {
    CURL *handle;
    cobalt::channel<int> *reponse_channel{nullptr};
};

/// This is a simple class making a lockfree::spsc_queue awaitable. It's not
/// movable, since the spsc_queue isn't.
template <typename T, typename... Options> struct awaitable_spsc_queue {
    boost::lockfree::spsc_queue<T, Options...> queue;

    template <typename... Args>
    awaitable_spsc_queue(Args &&...args) : queue(std::forward<Args>(args)...) {}

    // if the queue gets destroyed, destroying the awaiters is all we can do.
    ~awaitable_spsc_queue() {
        if (auto r = reader.load(); r != nullptr)
            std::coroutine_handle<void>::from_address(r).destroy();

        if (auto w = writer.load(); w != nullptr)
            std::coroutine_handle<void>::from_address(w).destroy();
    }
    // to avoid locks, put the coroutine handles into atomics.
    std::atomic<void *> reader{nullptr}, writer{nullptr};

    // capture the read & write executor
    cobalt::executor read_executor, write_executor;

    // the awaitable to read a value from the queue
    struct read_op {
        awaitable_spsc_queue *this_;

        // if reads are available we don't need to suspend
        bool await_ready() { return this_->queue.read_available(); }

        // The suspend implementation. We expect the coroutine promise to have
        // an associated executor. If it doesn't resumption will end up on the
        // system_executor
        template <typename Promise>
        void await_suspend(std::coroutine_handle<Promise> h) {
            // Capture the read_executor.
            if constexpr (requires { h.promise().get_executor(); })
                this_->read_executor = h.promise().get_executor();
            else
                this_->read_executor = cobalt::this_thread::get_executor();

            // Make sure there's only one coroutine awaiting the read
            assert(this_->reader == nullptr);
            // Store the handle of the awaiter.
            this_->reader.store(h.address());
        }
        T await_resume() {
            T res;
            // Grab the value from the queue
            this_->queue.pop(res);
            // if a writer is waiting post it to complete on it's thread
            auto w = cobalt::unique_handle<void>::from_address(
                this_->writer.exchange(nullptr));
            if (w)
                boost::asio::post(this_->write_executor, std::move(w));

            return res;
        }
    };

    // The function to be used with for reading.
    read_op read() { return {this}; }

    // the awaitable to write a value from the queue
    struct write_op {
        awaitable_spsc_queue *this_;
        // The value to write
        T value;

        // If there is room in the queue we don't need to suspend
        bool await_ready() { return this_->queue.write_available(); }

        // The suspend implementation. Similar to the writer one.
        template <typename Promise>
        void await_suspend(std::coroutine_handle<Promise> h) {
            // Capture the read_executor.
            if constexpr (requires { h.promise().get_executor(); })
                this_->write_executor = h.promise().get_executor();
            else
                this_->write_executor = cobalt::this_thread::get_executor();

            assert(this_->writer == nullptr);
            this_->writer.store(h.address());
        }

        // Resume the coroutine, this is where we do the actual write.
        // The reason is that we suspend when we cannot write because the queue
        // is full.
        void await_resume() {
            this_->queue.push(std::move(value));
            // if a writer is waiting post it
            auto r = cobalt::unique_handle<void>::from_address(
                this_->reader.exchange(nullptr));
            if (r)
                boost::asio::post(this_->read_executor, std::move(r));
        }
    };

    // The function to be used for writing.
    write_op write(T value) { return {this, std::move(value)}; }
};

class GlobalCurlService {
  public:
    static GlobalCurlService &Inst() {
        static GlobalCurlService inst;
        return inst;
    }
    cobalt::task<void> PushRequest(request &r) {
        co_await request_queue.write(r);
        bool test_val = false;
        if (loop_running.compare_exchange_strong(test_val, true)) {
            auto t = curl_request_loop();
            t.detach();
        }
    }

    std::atomic_bool loop_running{false};

  private:
    GlobalCurlService() : request_queue(1024) {}
    ~GlobalCurlService() {}
    awaitable_spsc_queue<request> request_queue;

    cobalt::thread curl_request_loop() {
        CURLM *multi = curl_multi_init();
        int max_handles = 1024;
        int running_handles = 0;
        auto add_requests = [&](int count) {
            request r;
            while (count > 0 && request_queue.queue.pop(r)) {
                curl_multi_add_handle(multi, r.handle);
                count--;
            }
        };
        int idle_loop_times = 0;
        const int max_idle_loop_times = 20;
        while (loop_running) {
            if (running_handles == 0 && request_queue.queue.empty()) {
                loop_running.exchange(idle_loop_times < max_idle_loop_times);
                std::this_thread::sleep_for(std::chrono::milliseconds(100));
                idle_loop_times++;
                continue;
            } else if (running_handles < max_handles) {
                add_requests(max_handles - running_handles);
            }
            idle_loop_times = 0;
            cobalt::channel<int> *ch = nullptr;
            CURLMcode mc = curl_multi_perform(multi, &running_handles);

            if (mc) {
                continue;
            }
            if (!mc && running_handles) {
                mc = curl_multi_poll(multi, NULL, 0, 1000, NULL);
            }
            if (mc) {
                continue;
            }
            struct CURLMsg *m;
            do {
                int msgq = 0;
                m = curl_multi_info_read(multi, &msgq);
                if (m && (m->msg == CURLMSG_DONE)) {
                    CURL *e = m->easy_handle;
                    long code;
                    curl_easy_getinfo(e, CURLINFO_HTTP_CODE, &code);
                    curl_easy_getinfo(e, CURLINFO_PRIVATE, &ch);
                    curl_multi_remove_handle(multi, e);
                    if (ch)
                        co_await ch->write((int)m->data.result);
                }

            } while (m);
        }
        curl_multi_cleanup(multi);
    }
};

cobalt::task<int> http_get(const std::string url) {
    cobalt::channel<int> ch_res;
    request r;
    r.reponse_channel = &ch_res;
    r.handle = curl_easy_init();
    curl_easy_setopt(r.handle, CURLOPT_URL, url.c_str());
    curl_easy_setopt(r.handle, CURLOPT_SSL_VERIFYPEER, 0);
    curl_easy_setopt(r.handle, CURLOPT_SSL_VERIFYHOST, 0);
    curl_easy_setopt(r.handle, CURLOPT_PRIVATE, &ch_res);
    co_await GlobalCurlService::Inst().PushRequest(r);
    auto rs = co_await ch_res.read();
    curl_easy_cleanup(r.handle);
    co_return rs;
}

cobalt::task<int> cc() {
    auto res = co_await http_get("https://www.baidu.com");
    std::cout << "\nhttp_get:" << res << std::endl;
    std::this_thread::sleep_for(std::chrono::seconds(2));
    std::cout << "wakeup" << std::endl;
    co_return 0;
}

// int main(int argc, char *argv[]) {
//     curl_global_init(CURL_GLOBAL_ALL);
//     // auto ret = cobalt::run(cc());
//     boost::asio::io_context ctx;
//     cobalt::this_thread::set_executor(ctx.get_executor());
//     cobalt::spawn(ctx, cc(), boost::asio::use_future);
//     ctx.run();
//     return 0;
// }